package com.atguigu.edu.realtime.dwd.db.split.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.bean.TableProcessDwd;
import com.atguigu.edu.realtime.common.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.*;

/**
 * @author Felix
 * @date 2024/11/04
 * 事实表数据动态分流处理
 */
public class BaseDbTableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>> {
    private MapStateDescriptor<String, TableProcessDwd> mapStateDescriptor;
    // TODO: 2024/11/14 解决主流数据先到 广播流数据 后到的 问题

    // TODO: 2024/11/14 mapstate
    private Map<String, TableProcessDwd> configMap = new HashMap<>();

    // TODO: 2024/11/14  mapstate 初始化,赋值 为 配置广播流中的数据
    public BaseDbTableProcessFunction(MapStateDescriptor<String, TableProcessDwd> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //将配置表中的配置信息预加载到程序中，解决主流数据先到，广播流数据后到的情况
    @Override
    public void open(Configuration parameters) throws Exception {
        Connection mySqlConnection = JdbcUtil.getMySqlConnection();
        String sql = "select * from edu_config.table_process_dwd";
        List<TableProcessDwd> tableProcessDwdList = JdbcUtil.queryList(mySqlConnection, sql, TableProcessDwd.class, true);
        for (TableProcessDwd tableProcessDwd : tableProcessDwdList) {
            String sourceTable = tableProcessDwd.getSourceTable();
            String sourceType = tableProcessDwd.getSourceType();
            String key = getKey(sourceTable, sourceType);
            configMap.put(key,tableProcessDwd);
        }
        JdbcUtil.closeMySqlConnection(mySqlConnection);
    }

    public String getKey(String sourceTable, String sourceType) {
        String key = sourceTable + ":" + sourceType;
        return key;
    }

    //processElement:处理主流业务数据   根据表名以及操作类型判断是否是需要动态分流处理的事实表数据，如果是，封装为Tuple2向下游传递
    @Override
    public void processElement(JSONObject jsonObj, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcessDwd>> out) throws Exception {
        //获取当前处理的业务数据库表表名
        String table = jsonObj.getString("table");
        //获取当前处理的业务数据库表操作类型
        String type = jsonObj.getString("type");
        //拼接key
        String key = getKey(table, type);
        //获取广播状态状态
        ReadOnlyBroadcastState<String, TableProcessDwd> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //根据key到广播状态以及configMap中获取对应的配置信息
        TableProcessDwd tableProcessDwd = null;
        if ((tableProcessDwd = broadcastState.get(key)) != null
          || (tableProcessDwd = configMap.get(key)) != null){
            //如果配置信息不为空，说明是需要动态分流处理的事实表数据  将其封装为Tuple2向下游传递
            //获取data部分内容---操作影响的一条数据
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");
            //在向下游传递数据前，过滤掉不需要传递的字段
            String sinkColumns = tableProcessDwd.getSinkColumns();
            deleteNotNeedColumns(dataJsonObj,sinkColumns);
            //在向下游传递数据前，补充事件时间字段
            Long ts = jsonObj.getLong("ts");
            dataJsonObj.put("ts",ts);
            out.collect(Tuple2.of(dataJsonObj,tableProcessDwd));
        }
    }

    private void deleteNotNeedColumns(JSONObject dataJsonObj, String sinkColumns) {
        List<String> columnList = Arrays.asList(sinkColumns.split(","));
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry->!columnList.contains(entry.getKey()));
    }

    //processBroadcastElement:处理广播流数据  将配置表中的配置信息放到广播状态状态以及configMap或者从广播状态以及configMap删除
    @Override
    public void processBroadcastElement(TableProcessDwd tableProcessDwd, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.Context ctx, Collector<Tuple2<JSONObject, TableProcessDwd>> out) throws Exception {
        //获取对配置表进行的操作的类型
        String op = tableProcessDwd.getOp();
        //获取广播状态
        BroadcastState<String, TableProcessDwd> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //拼接key
        String sourceTable = tableProcessDwd.getSourceTable();
        String sourceType = tableProcessDwd.getSourceType();
        String key = getKey(sourceTable, sourceType);

        if("d".equals(op)){
            //说明从配置表中删除了一条配置  需要将对应的配置信息也从广播状态以及configMap中删除掉
            broadcastState.remove(key);
            configMap.remove(key);
        }else {
            //说明对配置表中进行了读取、更新以及添加  需要将对应的配置信息也更新到广播状态以及configMap中
            broadcastState.put(key,tableProcessDwd);
            configMap.put(key,tableProcessDwd);
        }

    }
}
